use crate::signaling::{close_and_remove, message::SignalMessage, G01Client};
use futures_util::{SinkExt, StreamExt, TryStreamExt};
use rand::Rng;
use reqwest::Client;
use reqwest_websocket::{Message, RequestBuilderExt};
use serde::{Deserialize, Serialize};
use std::{
    collections::HashMap,
    sync::{Arc, RwLock},
};
use tokio::time::{sleep, timeout, Duration};

const CONNECTION: &'static str = "wss://www.idns.link/ws";
// const CONNECTION: &'static str = "ws://127.0.0.1:7070";

lazy_static::lazy_static! {
    //
    pub(crate) static ref RESPONSE_BUFFER: RwLock<HashMap<u64, Option<SignalMessage>>> = RwLock::new(HashMap::new());
}

pub async fn connect_g01_signaling(token: &String) -> anyhow::Result<()> {
    if super::signaling_is_login()? {
        tracing::debug!("已连接！");
        return Ok(());
    }
    //获取节点配置
    let peer_config = model_graph_common::config::get_peer_config().await?;
    tracing::debug!("peer_id:{}", peer_config.peer_id);
    //拼接地址
    let url = format!(
        "{}?authorization={}&peer_id={}",
        CONNECTION, token, peer_config.peer_id
    );

    let (sender_tx, mut sender_rx) = tokio::sync::mpsc::unbounded_channel::<String>();

    let websocket = Client::default()
        .get(url)
        .upgrade()
        .send()
        .await?
        .into_websocket()
        .await?;

    let (mut tx, mut rx) = websocket.split();

    let (done_tx, mut done_rx) = tokio::sync::mpsc::unbounded_channel::<()>();

    let token_1 = token.clone();

    tokio::spawn(async move {
        loop {
            tokio::select! {
                msg = rx.try_next()=>{
                    match msg {
                        Ok(Some(message))=>{
                            match message {
                                Message::Text(data) => {
                                    tracing::debug!("recv:{}", data);
                                    if data == r#"{"type":"pong"}"#{

                                    }else{
                                        //
                                        let msg: anyhow::Result<SignalMessage> =
                                            serde_json::from_str(&data).map_err(|err| anyhow::anyhow!("{}", err));
                                        match msg {
                                            Ok(msg) => {
                                                //如果还有request_id那么
                                                if let Ok(msg_option) = put_response(msg){
                                                    if let Some(msg) = msg_option{
                                                        tracing::debug!("消息处理:{:?}", msg);
                                                        let on_res = super::handler::on_message(msg).await;
                                                        match on_res {
                                                            Ok(_) => {
                                                                tracing::debug!("消息处理成功！");
                                                            }
                                                            Err(err) => {
                                                                tracing::error!("消息处理失败:{:?}", err);
                                                            }
                                                        }
                                                    }else{
                                                        //消息作为实时处理
                                                        tracing::debug!("request_response消息处理成功！");
                                                    }
                                                }else{
                                                    //失败
                                                    tracing::error!("消息处理失败!");
                                                }
                                            }
                                            Err(err) => {
                                                tracing::error!("JSON解析失败:{:?}", err);
                                            }
                                        }

                                    }
                                }
                                _ => {}
                            }
                        }
                        _=>{
                            //判断连接是否可以用
                            tracing::debug!("接收失败，关闭连接!");
                            break;
                        }
                    }
                }
            }
        }
        //重新登录
        tracing::debug!("关闭连接!");
        let _ = close_and_remove(token_1).await;
    });

    let token_1 = token.clone();

    tokio::spawn(async move {
        let done_tx_1 = done_tx.clone();

        loop {
            tokio::select! {
                _= done_rx.recv()=>{
                    tracing::debug!("关闭连接!");
                    break;
                }
                msg = sender_rx.recv() => {
                    match msg{
                        Some(msg)=>{
                            let result = tx.send(Message::Text(String::from(&msg))).await;
                            match result{
                                Ok(_)=>{
                                    tracing::debug!("发送成功!");
                                }
                                Err(err)=>{
                                    tracing::debug!("发送信令失败!{}",err);
                                   let _ = done_tx_1.send(());
                                }
                            }
                        }
                        None=>{
                            tracing::debug!("没有接收到消息!");
                            break;
                        }
                    }
                }
            }
        }
        //重新登录
        tracing::debug!("关闭连接!");
        let _ = close_and_remove(token_1).await;
    });

    //
    {
        let mut client = super::SIGNALING_CLIENT
            .write()
            .map_err(|err| anyhow::anyhow!("获取Client失败!{}", err))?;

        *client = Some(G01Client { sender: sender_tx });
    }

    Ok(())
}

pub async fn send_string(data: &String) -> anyhow::Result<()> {
    //
    let client = super::SIGNALING_CLIENT
        .read()
        .map_err(|err| anyhow::anyhow!("获取Client失败!{}", err))?;

    match client.as_ref() {
        Some(client) => {
            let _ = client.sender.clone().send(data.clone());
        }
        None => {
            tracing::error!("没有连接!");
        }
    }
    Ok(())
}

pub async fn request_response<T>(
    type_str: &String,
    data: T,
    timeout_millis: u64,
) -> anyhow::Result<SignalMessage>
where
    T: Serialize,
{
    //获取sender
    tracing::debug!("开始请求.....",);

    if let Ok(response_result) = timeout(
        Duration::from_millis(timeout_millis),
        Box::pin(async move {
            //发送结果
            _response(type_str, data).await
        }),
    )
    .await
    {
        let response = response_result.map_err(|err| anyhow::anyhow!("{}", err))?;
        tracing::debug!("返回数据成功!");
        return Ok(response);
    } else {
        tracing::debug!("请求超时!");
        return Err(anyhow::anyhow!("请求超时!"));
    }
}

async fn _response<T>(type_str: &String, data: T) -> anyhow::Result<SignalMessage>
where
    T: Serialize,
{
    //设置一个
    let request_id = generate_push_request_id()?;
    tracing::debug!("请求ID:{}", request_id);

    //
    let _ = {
        let client = super::SIGNALING_CLIENT
            .read()
            .map_err(|err| anyhow::anyhow!("获取Client失败!{}", err))?;

        match client.as_ref() {
            Some(client) => {
                let _ = client.sender.clone().send(
                    serde_json::to_string(&serde_json::json!({
                        "type": type_str,
                        "data": data,
                        "request_id":request_id,
                    }))
                    .map_err(|err| anyhow::anyhow!(""))?,
                );
            }
            None => {
                tracing::error!("没有连接!");
                return Err(anyhow::anyhow!("没有获取到连接!"));
            }
        }
    };

    //发送成功,循环进行等待结果
    loop {
        let pop_res = pop_response(request_id).await;
        match pop_res {
            Ok(res_message) => {
                if let Some(response) = res_message {
                    //获取到返回值
                    return Ok(response);
                } else {
                    //
                    sleep(Duration::from_millis(10)).await;
                }
            }
            Err(err) => {
                let _ = remove_response(request_id).await;

                return Err(anyhow::anyhow!("获取结果失败{}!", err));
            }
        }
    }
}

fn push_response(request_id: u64) -> anyhow::Result<bool> {
    //判断是否存在
    let exist = {
        let cache = RESPONSE_BUFFER
            .read()
            .map_err(|err| anyhow::anyhow!("{}", err))?;
        cache.contains_key(&request_id)
    };
    if exist {
        Ok(false)
    } else {
        //
        let mut cache_result = RESPONSE_BUFFER
            .write()
            .map_err(|err| anyhow::anyhow!("获取锁失败:{}", err))?;

        cache_result.insert(request_id, None);
        tracing::debug!("插入成功!");
        Ok(true)
    }
}

fn generate_push_request_id() -> anyhow::Result<u64> {
    //随机一个值
    let mut rng = rand::thread_rng();

    let mut request_id: u64 = rng.gen();
    while !push_response(request_id)? {
        request_id = rng.gen();
    }
    Ok(request_id)
}

async fn pop_response(request_id: u64) -> anyhow::Result<Option<SignalMessage>> {
    //
    let mut cache_result = RESPONSE_BUFFER
        .write()
        .map_err(|err| anyhow::anyhow!("获取锁失败:{}", err))?;
    if let Some(message) = cache_result.get(&request_id) {
        if message.is_some() {
            if let Some(message) = cache_result.remove(&request_id) {
                Ok(message)
            } else {
                Err(anyhow::anyhow!("不存在该请求id!{}", request_id))
            }
        } else {
            Ok(None)
        }
    } else {
        Err(anyhow::anyhow!("不存在该请求id!{}", request_id))
    }
}

async fn remove_response(request_id: u64) -> anyhow::Result<Option<SignalMessage>> {
    //
    let mut cache_result = RESPONSE_BUFFER
        .write()
        .map_err(|err| anyhow::anyhow!("获取锁失败:{}", err))?;
    if let Some(message) = cache_result.remove(&request_id) {
        Ok(message)
    } else {
        Ok(None)
    }
}

fn put_response(response: SignalMessage) -> anyhow::Result<Option<SignalMessage>> {
    if let Some(request_id) = response.request_id {
        //
        let mut cache_result = RESPONSE_BUFFER
            .write()
            .map_err(|err| anyhow::anyhow!("获取锁失败:{}", err))?;

        if let Some(message) = cache_result.get_mut(&request_id) {
            *message = Some(response);
            Ok(None)
        } else {
            Ok(Some(response))
        }
    } else {
        Ok(Some(response))
    }
}
